package cn.classg.rabbitmq.config;

import cn.classg.rabbitmq.annotation.MqBinding;
import cn.classg.rabbitmq.annotation.MqExchange;
import cn.classg.rabbitmq.annotation.RabbitSettingProducer;
import cn.classg.rabbitmq.callback.AbstractMqCallback;
import cn.classg.rabbitmq.callback.RabbitTemplateCallback;
import cn.classg.rabbitmq.context.MqSpringContextUtil;
import cn.classg.rabbitmq.util.JdkSerialization;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.lang.reflect.Type;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息代理
 * @author CLASSG
 */
@Slf4j
public class MessagePorducerInvocationHandler extends RabbitTemplate implements InvocationHandler {

    /**
     * 缓存mq 对应的注解
     */
    private static final Map<Class<?>, RabbitSettingProducer> MESSAGE_QUEUE_MAP = new ConcurrentHashMap<>();



    private static boolean callbackFlag = false;


    public static Object generateProxyClass(Class<?> type, boolean callbackFlag) {
        MessagePorducerInvocationHandler.callbackFlag = callbackFlag;
        return Proxy.newProxyInstance(type.getClassLoader(), new Class[]{type}, new MessagePorducerInvocationHandler());
    }


    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        if (method.getDeclaringClass() == Object.class) {
            return method.invoke(this, args);
        }
        RabbitSettingProducer mqProducer = getMessageQueue(proxy);
        MqBinding bindings = mqProducer.bindings();
        if (Objects.isNull(bindings)) {
            log.warn("没有配置binding，发送出错");
            return null;
        }

        Object body;
        MessagePostProcessor messagePostProcessor = null;
        if (args.length == 1) {
            body = args[0];
        } else if (args.length == 2) {
            body = args[0];
            messagePostProcessor = (MessagePostProcessor) args[1];
        } else {
            log.error("mq 发送失败，代理方法有问题");
            return null;
        }
        CorrelationData data = new CorrelationData(UUID.randomUUID().toString());

        // open callback function
        if (MessagePorducerInvocationHandler.callbackFlag) {
            Class<? extends AbstractMqCallback> callback = mqProducer.callback();
            if (!callback.getName().equals(AbstractMqCallback.class.getName())) {
                data.setReturnedMessage(new Message(JdkSerialization.serialize(body)));
                RabbitTemplateCallback.addCorrelationDataToTL(data.getId(), callback,body.getClass());
            }
        }
        MqExchange exchange = bindings.exchange();
        String key = bindings.key();

        RabbitTemplate rabbitTemplate = MqSpringContextUtil.getBean(RabbitTemplate.class);
        if (exchange.type().equals(ExchangeTypes.DIRECT)) {
            switch (args.length) {
                case 1:
                    rabbitTemplate.convertAndSend(MqSpringContextUtil.getSpringValue(exchange.value()), key, body, data);
                    break;
                case 2:
                    rabbitTemplate.convertAndSend(MqSpringContextUtil.getSpringValue(exchange.value()), key, body, messagePostProcessor, data);
                    break;
                default:
            }

        } else if (exchange.type().equals(ExchangeTypes.FANOUT)) {
            switch (args.length) {
                case 1:
                    rabbitTemplate.convertAndSend(MqSpringContextUtil.getSpringValue(exchange.value()), StringUtils.EMPTY, body, data);
                    break;
                case 2:
                    rabbitTemplate.convertAndSend(MqSpringContextUtil.getSpringValue(exchange.value()), StringUtils.EMPTY, body, messagePostProcessor, data);
                    break;
                default:
            }
        } else {
            log.error("mq 暂不支持此路由类型");
            return null;
        }

        return null;
    }

    private RabbitSettingProducer getMessageQueue(Object proxy) {
        RabbitSettingProducer messageQueue;
        if ((messageQueue = MESSAGE_QUEUE_MAP.get(proxy.getClass())) != null) {
            return messageQueue;
        }
        Type[] genericInterfaces = proxy.getClass().getGenericInterfaces();
        Class<?> genericInterface = (Class<?>) genericInterfaces[0];
        RabbitSettingProducer annotation = genericInterface.getAnnotation(RabbitSettingProducer.class);
        MESSAGE_QUEUE_MAP.put(proxy.getClass(), annotation);
        return annotation;

    }

}
